package com.zbkj.front.service.mqtt;

import com.zbkj.common.config.AliMqttConfig;
import com.zbkj.common.utils.mqtt.ConnectionOptionWrapper;
import com.zbkj.common.utils.mqtt.MqttTools;
import com.zbkj.front.service.MqttClientService;
import com.zbkj.service.service.ZfbMonitorDeviceBindService;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.event.EventListener;
import org.springframework.core.env.Environment;
import org.springframework.stereotype.Service;

import java.security.InvalidKeyException;
import java.security.NoSuchAlgorithmException;
import java.util.concurrent.*;

/**
 * MQTT消费服务类
 */
@Service
@Slf4j
public class MqttClientServiceImpl implements MqttClientService {

    @Autowired
    private AliMqttConfig aliMqttConfig;

    @Autowired
    private ZfbMonitorDeviceBindService monitorDeviceBindService;
    @Autowired
    private Environment env;

    @Override
    @EventListener(ApplicationReadyEvent.class)
    public void subscribe() throws MqttException {
        String port = env.getProperty("local.server.port", "默认端口号");
        log.info("端口号port = " + port);
        if (!port.equals("20710")) {
            log.info("其他实例，不启动MQTT消费者");
            return;
        }
//        String osName = System.getProperty("os.name").toLowerCase();
        // 本地测试
//        if (!osName.contains("windows")) {
//            return;
//        }
         int THREAD_POOL_SIZE = 5; // 根据任务量调整
         BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>(1000);
        ExecutorService executorService = new ThreadPoolExecutor(
                THREAD_POOL_SIZE,
                THREAD_POOL_SIZE,
                0L, TimeUnit.MILLISECONDS,
                workQueue
        );

        MemoryPersistence memoryPersistence = new MemoryPersistence();
        MqttClient mqttClient = new MqttClient("tcp://" + aliMqttConfig.serverUrl, aliMqttConfig.clientId2, memoryPersistence);
        mqttClient.setTimeToWait(5000);
        // 设置MQTT回调函数
        mqttClient.setCallback(new MqttCallbackExtended() {
            @Override
            public void connectComplete(boolean reconnect, String serverURI) {
                log.info("可视认养，启动MQTT结果消费者");
            }

            @Override
            public void connectionLost(Throwable throwable) {
                log.info("可视认养，接收器,MQTT链接断开，将会重连");
            }

            @Override
            public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
                String msg = new String(mqttMessage.getPayload());
                // 异步提交给线程池处理数据库操作
                executorService.submit(() -> processMessage(topic, msg));
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
                // 消息发送完成后的处理逻辑
                log.info("可视认养，接收器send msg succeed topic is : " + iMqttDeliveryToken.getTopics()[0]);
            }
        });
        // 连接MQTT
        MqttConnectOptions connOpts = aliMqttConfig.connectionOptionWrapper.getMqttConnectOptions();
        connOpts.setAutomaticReconnect(true);
        connOpts.setKeepAliveInterval(60); // 设置心跳间隔为60秒
        connOpts.setConnectionTimeout(30); // 设置连接超时时间为30秒
        mqttClient.connect(connOpts);
        // 订阅MQTT，一旦有动物（BF29小椭圆设备）靠近网关时就会推送消息到这个主题的
        String mq4IotTopic = aliMqttConfig.parentTopic + "/bf29/" + "clientdata";
        mqttClient.subscribe(mq4IotTopic, 0);
    }

    private void processMessage(String topic, String msg) {
        try {
            if (topic.endsWith("clientdata")) {
                log.info("接收消息Received: " + msg + ",topic=" + topic);
                String[] split = msg.split("\\+");
                String imei = split[0].replace("mac:", "");
                String mac = split[1]; // mac地址，带冒号:
                // 更新绑定的设备表，更新设备时间，要想知道牛离哪个设备近，就用更新时间排序可知
                monitorDeviceBindService.updateRefreshTime(imei, mac);
            } else if (topic.endsWith("status")) {
                log.info("接收状态消息Received: " + msg + ",topic=" + topic);
                if (msg.contains("+")) {
                    String[] split = msg.split("\\+");
                    String imei = split[0];
                    String mac = split[1];
                    String res = split[2];
                    // 格式：CE:04:40:06:00:10
                    String outMac = MqttTools.formatMacWithColonRegex(mac);
                    monitorDeviceBindService.updateStatus(imei, outMac, res);
                }
            }
        } catch (Exception e) {
            Thread.currentThread().interrupt();
            e.printStackTrace();
        }
    }


//    @Override
    @EventListener(ApplicationReadyEvent.class)
    public void subscribeDeviceStatus() throws MqttException {
        String port = env.getProperty("local.server.port", "默认端口号");
        log.info("端口号port = " + port);
        if (!port.equals("20710")) {
            log.info("其他实例，不启动MQTT消费者");
            return;
        }
//        String osName = System.getProperty("os.name").toLowerCase();
        // 本地测试
//        if (!osName.contains("windows")) {
//            return;
//        }
        int THREAD_POOL_SIZE = 5; // 根据任务量调整
        BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>(1000);
        ExecutorService executorService = new ThreadPoolExecutor(
                THREAD_POOL_SIZE,
                THREAD_POOL_SIZE,
                0L, TimeUnit.MILLISECONDS,
                workQueue
        );

        MemoryPersistence memoryPersistence = new MemoryPersistence();
        MqttClient client = new MqttClient("tcp://" + aliMqttConfig.serverUrl, aliMqttConfig.clientId6, memoryPersistence);
        client.setTimeToWait(5000);
        client.setCallback(new MqttCallbackExtended() {
            @Override
            public void connectComplete(boolean reconnect, String serverURI) {
                log.info("启动MQTT状态消费者");
            }

            @Override
            public void connectionLost(Throwable throwable) {
                log.info("STATUS,MQTT链接断开，将会重连");
            }

            @Override
            public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
                log.info("STATUS,接收消息【" + topic + "】 , 内容【" + new String(mqttMessage.getPayload()) + "】");
                String msg = new String(mqttMessage.getPayload());
                // 异步提交给线程池处理数据库操作
                executorService.submit(() -> processMessage(topic, msg));
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
                // 消息发送完成后的处理逻辑
                log.info("STATUS,send msg succeed topic is : " + iMqttDeliveryToken.getTopics()[0]);
            }
        });
        try {
            ConnectionOptionWrapper connectionOptionWrapper = new ConnectionOptionWrapper(aliMqttConfig.instanceId, aliMqttConfig.accessKeyId, aliMqttConfig.accessKeySecret, aliMqttConfig.clientId6);
            client.connect(connectionOptionWrapper.getMqttConnectOptions());
        } catch (NoSuchAlgorithmException e) {
            throw new RuntimeException(e);
        } catch (InvalidKeyException e) {
            throw new RuntimeException(e);
        }
        // 订阅MQTT状态，当发送指令给MQTT后，BF29设备没有亮灯时返回FAIL，成功返回OK
        String mq4IotTopic = aliMqttConfig.parentTopic + "/bf29/" + "status";
        client.subscribe(mq4IotTopic, 0);

//        client.subscribe(mq4IotTopic, (topic, message) -> {
//            String msg = new String(message.getPayload());
//            log.info("STATUS,订阅MQTT状态Received Status: " + msg);
//            String[] split = msg.split("\\+");
//            String imei = split[0];
//            String mac = split[1];
//            String res = split[2];
//            // mac补充:
//            String outMac = MqttTools.formatMacWithColonRegex(mac);
//            log.info("转换后Mac = " + outMac);
//            monitorDeviceBindService.updateStatus(imei, outMac, res);
//        });
    }

}
